package com.zhou.common.lib.serilizer;

import com.esotericsoftware.kryo.kryo5.Kryo;
import com.esotericsoftware.kryo.kryo5.io.Input;
import com.esotericsoftware.kryo.kryo5.io.Output;
import com.esotericsoftware.kryo.kryo5.util.Pool;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;

/**
 * @author li.ke
 * @date 2021/12/7
 */
public class KryoSerializer<T> implements RedisSerializer<T> {
    private static final Pool<Kryo> kryoPool = new Pool<Kryo>(true, false, 8) {
        @Override
        protected Kryo create() {
            Kryo kryo = new Kryo();
            kryo.setRegistrationRequired(false);
            kryo.setReferences(true);
            return kryo;
        }
    };

    private static final Pool<Output> outputPool = new Pool<Output>(true, false, 8) {
        @Override
        protected Output create() {
            Output output = new Output(4096, 1024 * 1024 * 5);
            return output;
        }
    };

    private static final Pool<Input> inputPool = new Pool<Input>(true, false, 8) {
        @Override
        protected Input create() {
            Input input = new Input(4096);
            return input;
        }
    };

    public byte[] serialize(T obj) throws SerializationException {
        Kryo kryo = null;
        Output output = null;
        try {
            output = outputPool.obtain();
            kryo = kryoPool.obtain();
            kryo.writeClassAndObject(output, obj);
            output.close();

            return output.toBytes();
        } catch (Exception ex) {
            throw new SerializationException(ex.getMessage());
        } finally {
            if (kryo != null) {
                kryoPool.free(kryo);
            }

            if (output != null) {
                outputPool.free(output);
            }
        }
    }

    @Override
    public T deserialize(byte[] content) throws SerializationException {
        Kryo kryo = null;
        Input input = null;
        try {
            input = inputPool.obtain();
            input.setBuffer(content);
            kryo = kryoPool.obtain();
            Object obj = kryo.readClassAndObject(input);
            return (T) obj;
        } catch (Exception ex) {
            throw new SerializationException(ex.getMessage());
        } finally {
            if (kryo != null) {
                kryoPool.free(kryo);
            }

            if (input != null) {
                inputPool.free(input);
            }
        }
    }
}
